package cn.monkey.transport.mq.rabbitmq;

import cn.monkey.transport.core.Exchange;
import com.rabbitmq.client.ConnectionFactory;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Scheduler;
import reactor.rabbitmq.*;

import java.io.IOException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeoutException;

public class SimpleRabbitMqExchangeFactory extends AbstractRabbitMqExchangeFactory<RabbitMqExchange> {

    private ConnectionFactory connectionFactory = new ConnectionFactory();
    private Scheduler scheduler;

    public void setConnectionFactory(ConnectionFactory connectionFactory) {
        connectionFactory.useNio();
        this.connectionFactory = connectionFactory;
    }

    public void setScheduler(Scheduler scheduler) {
        this.scheduler = scheduler;
    }

    @Override
    public RabbitMqExchange apply(String key, Exchange exchange, RabbitMqTransportConfig rabbitMqTransportConfig) {
        RabbitMqOutbound rabbitMqOutbound = this.newOutbound(rabbitMqTransportConfig.getOutbound());
        RabbitMqInbound rabbitMqInbound = this.newInbound(rabbitMqTransportConfig.getInbound());
        return new SimpleRabbitMqExchange(key, rabbitMqInbound, rabbitMqOutbound);
    }

    @Override
    protected RabbitMqInbound newInbound(RabbitMqTransportConfig.Inbound inbound) {
        ReceiverOptions receiverOptions = new ReceiverOptions()
                // do not use #connectionSupplier here,
                // case receiver#close will block execute
                .connectionMono(Mono.create(sink -> CompletableFuture.runAsync(() -> {
                    try {
                        sink.success(connectionFactory.newConnection(inbound.getAddresses()));
                    } catch (IOException | TimeoutException e) {
                        sink.error(e);
                    }
                })))
                .connectionSubscriptionScheduler(this.scheduler);
        Receiver receiver = RabbitFlux.createReceiver(receiverOptions);
        return new SimpleRabbitMqInbound(receiver, inbound.getCodec(), inbound.getQueue(), inbound.getConsumeOptions());
    }

    @Override
    protected RabbitMqExchange inverse(RabbitMqExchange exchange) {
        throw new UnsupportedOperationException();
    }

    @Override
    protected RabbitMqOutbound newOutbound(RabbitMqTransportConfig.Outbound outbound) {
        SenderOptions senderOptions = new SenderOptions()
                .connectionMono(Mono.create(sink -> CompletableFuture.runAsync(() -> {
                    try {
                        sink.success(connectionFactory.newConnection(outbound.getAddresses()));
                    } catch (IOException | TimeoutException e) {
                        sink.error(e);
                    }
                })))
                .resourceManagementScheduler(this.scheduler);
        Sender sender = RabbitFlux.createSender(senderOptions);
        return new SimpleRabbitMqOutbound(sender, outbound.getCodec(), outbound.getExchange(), outbound.getRouterKey(), outbound.getSendOptions(), outbound.getCacheSize());
    }
}
